#   todo 本文件为数据的处理
import pandas  as pd
from hdfs.client import Client
import pymysql

#   todo 修改pandas的内置参数，让其显示所有的行和列
#   显示所有列数据
pd.set_option('display.max_columns',None)
#   显示所有行数据
pd.set_option('display.max_rows',None)
#   设置value的长度
pd.set_option('max_colwidth',100)
#   设置1000列时才换行
pd.set_option('display.width',1000)


data=pd.read_csv('source.csv')


#   todo 1.去除重复数据
# print("去重前的数据量:",len(data))
data.drop_duplicates(inplace=True)
# print("去重后的数据量:",len(data))

#   todo 2.去除空值行
print("去除空行前:",len(data))
data.dropna(inplace=True)
print("去除空行后:",len(data))

#   todo 3.国家标准化
print(data)
data['guojia']=data['guojia'].apply(
    lambda x: "中国" if x in ["中国大陆","中国台湾","中国香港","中国香港 中国大陆","中国大陆 中国香港","中国香港 中国台湾"]  else x
)
print(data['guojia'])

#   todo 4.添加列 添加一列，判断是否有中国演员参演
data['china']=data['guojia'].apply(lambda x: 1 if "中国" in x else 0)
print(data['china'])


#   todo 修改列
print(data['guojia'])
data['guojia']=data['guojia'].apply(lambda x: "多个国家" if len(x.split(" "))!=1 else x)
print(data['guojia'])

#   todo 处理完的数据存入本地
data.to_csv('clean.csv',index=False,header=True)


#   todo 下面将数据写入mysql和hdfs
#   todo 把数据写入到hdfs
#   创建一个hdfs的客户端实例(9870是web访问hdfs的端口号)
client=Client('http://192.168.40.110:9870')
#   hdfs上面存储的路径
hdfs_path="/fyp/"
#   使用客户端的upload方法将文件上传到hdfs
client.upload(hdfs_path=hdfs_path,local_path='clean.csv',overwrite=True)
print("处理完的数据已经保存到hdfs,保存的路径为:/fyp/clean.csv........................")


#     将处理好的数据写入mysql
#   建立与mysql的连接
connect=pymysql.connect(host='192.168.40.110',user='root',password='123456',database='fyp',charset='utf8')
#   通过连接创建游标对象
cursor=connect.cursor()
#   定义sql如果存在表就删除表(data_processing:数据处理)
drop_table_sql='drop table if exists clean;'
cursor.execute(drop_table_sql)
#   定义创建表格的sql
CreateTable_sql="""
create table if not exists clean(
            id int primary key AUTO_INCREMENT,
            name varchar(500),
            year varchar(500),
            guojia varchar(1000),
            class varchar(500),
            director varchar(500),
            zhujue varchar(500),
            mark double,
            image_url varchar(500),
            info_url varchar(500),
            china int
);
"""
#   执行创建表的语句
cursor.execute(CreateTable_sql)
#   定义插入数据的sql语句
insert_sql="""
    insert into clean(
         name,year,guojia,class,director,zhujue,mark,image_url,info_url,china)
    values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);
"""
#   换成list[row]的形式方便写入mysql
data_list=data.values.tolist()
#   执行插入数据
cursor.executemany(insert_sql,data_list)

#   提交任务(只有提交了任务才会执行sql)
connect.commit()
#   关闭游标对象
cursor.close()
#   关闭连接
connect.close()
print("处理完的数据已经保存到mysql的fyp数据库的clean表..............")